package org.aurora.remote.netty;

import io.netty.channel.Channel;
import org.aurora.common.thread.ServiceThread;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

public class NettyEventPublisher extends ServiceThread {
    private static final Logger log = LoggerFactory.getLogger(NettyEventPublisher.class);
    private final List<NettyEventSubscriber> subscribers = new ArrayList<>();
    private final BlockingDeque<NettyEvent> nettyEventBlockingDeque = new LinkedBlockingDeque();

    public NettyEventPublisher() {
        super("nettyEventPublisher");
    }

    public void addSubscriber(NettyEventSubscriber subscriber) {
        this.subscribers.add(subscriber);
    }

    public void removeSubscriber(NettyEventSubscriber subscriber) {
        this.subscribers.remove(subscriber);
    }

    public void publishEvent(NettyEvent nettyEvent) {
        this.nettyEventBlockingDeque.offer(nettyEvent);
    }

    @Override
    public void run() {
        while (true) {
            try {
                if (this.isStop()) {
                    log.info("nettyEventPublisher stop");
                    return;
                }

                NettyEvent nettyEvent = nettyEventBlockingDeque.poll(3000, TimeUnit.MILLISECONDS);

                if (nettyEvent == null) {
                    continue;
                }

                Channel channel = nettyEvent.getChannel();

                InetSocketAddress inetSocketAddress = (InetSocketAddress) channel.remoteAddress();
                String ip = inetSocketAddress.getAddress().getHostAddress();
                int port = inetSocketAddress.getPort();
                nettyEvent.setIp(ip);
                nettyEvent.setPort(port);

                for (NettyEventSubscriber subscriber : subscribers) {
                    try {
                        subscriber.processorEvent(nettyEvent);
                    } catch (Exception e) {
                        log.error("subscriber execute error, subscriber name = %s ", subscriber != null ? subscriber.getSubscriberName() : "", e);
                    }
                }
            } catch (Exception e) {
                log.error("nettyEventPublisher run error", e);
            }
        }
    }
}
